package rpc.utils.concurrent.threadpool;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @ClassName: ThreadPoolFactoryUtils
 * @Description: 用于创建线程池的工具类
 * @author: Kowk
 * @since:1.0.0
 */
@Slf4j
@NoArgsConstructor
public class ThreadPoolFactoryUtils {
    //ExecutorService 会自动提供一个线程池和相关 API，用于为其分配任务。
    /**
     * 通过 threadNamePrefix 来区分不同线程池（我们可以把相同 threadNamePrefix 的线程池看作是为同一业务场景服务）。
     * key: threadNamePrefix
     * value: threadPool
     */
    private static final Map<String , ExecutorService> THREAD_POOLS = new ConcurrentHashMap<>();

    public static ExecutorService createCustomThreadPoolIfAbsent(String threadNamePrefix){
        CustomThreadPoolConfig customThreadPoolConfig = new CustomThreadPoolConfig();
        return createCustomThreadPoolIfAbsent(customThreadPoolConfig, threadNamePrefix, false);
    }
    //daemon表示是否为守护进程
    public static ExecutorService createCustomThreadPoolIfAbsent(CustomThreadPoolConfig customThreadPoolConfig, String threadNamePrefix, Boolean daemon){
        ExecutorService threadPool = THREAD_POOLS.computeIfAbsent(threadNamePrefix, k -> createThreadPool(customThreadPoolConfig, threadNamePrefix, daemon));
        //如果线程池被关闭就重新创建
        if(threadPool.isShutdown()||threadPool.isTerminated()){
            THREAD_POOLS.remove(threadNamePrefix);
            threadPool = createThreadPool(customThreadPoolConfig, threadNamePrefix, daemon);
            THREAD_POOLS.put(threadNamePrefix,threadPool);
        }
        return threadPool;
    }

    private static ExecutorService createThreadPool(CustomThreadPoolConfig customThreadPoolConfig, String threadNamePrefix, Boolean daemon) {
        ThreadFactory threadFactory = createThreadFactory(threadNamePrefix, daemon);
        return new ThreadPoolExecutor(customThreadPoolConfig.getCorePoolSize(), customThreadPoolConfig.getMaximumPoolSize(), customThreadPoolConfig.getKeepAliveTime(),customThreadPoolConfig.getUnit(),customThreadPoolConfig.getWorkQueue(),threadFactory);
    }

    /**
     * @Description: 创建一个线程工厂用于创建线程，如果前缀为空就是用默认的工厂类，否则就自己新建一个工厂类
     * @Param: [threadNamePrefix 线程名字的前缀, daemon 是否为守护线程]
     * @return: java.util.concurrent.ThreadFactory
     */
    public static ThreadFactory createThreadFactory(String threadNamePrefix, Boolean daemon) {
        if (threadNamePrefix != null) {
            if (daemon != null) {
                return new ThreadFactoryBuilder().setNameFormat(threadNamePrefix+"-%d")
                        .setDaemon(daemon)
                        .build();
            }
            else {
                return new ThreadFactoryBuilder().setNameFormat(threadNamePrefix+"-%d").build();
            }
        }
        return Executors.defaultThreadFactory();
    }

    /**
     * @Description: 输出线程池的状态
     * @Param: [threadPoolExecutor]
     * @return: void
     */
    public static void printThreadPoolStatus(ThreadPoolExecutor threadPoolExecutor){

        ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1, createThreadFactory("print-thread-pool-status",false));
        scheduledExecutorService.scheduleAtFixedRate(()->{
            log.info("============ThreadPool Status=============");
            log.info("ThreadPool Size: [{}]", threadPoolExecutor.getPoolSize());
            log.info("Active Threads: [{}]", threadPoolExecutor.getActiveCount());
            log.info("Number of Tasks : [{}]", threadPoolExecutor.getCompletedTaskCount());
            log.info("Number of Tasks in Queue: {}", threadPoolExecutor.getQueue().size());
            log.info("===========================================");
        },0,1,TimeUnit.SECONDS);
    }

    public static void shutDownAllThreadPool(){
        log.info("开始关闭所有线程池");
        //parallelStream是一个并行执行的流.它通过默认的ForkJoinPool,可能提高你的多线程任务的速度. 一般不会按顺序去遍历元素除非是用forEachOrder
        THREAD_POOLS.entrySet().parallelStream().forEach(entry -> {
            ExecutorService executorService = entry.getValue();
            executorService.shutdown();
            log.info("关闭线程池[{}]-[{}]",entry.getKey(),executorService.isTerminated());
            try {
                executorService.awaitTermination(10,TimeUnit.SECONDS);

            } catch (InterruptedException e) {
                log.error("线程池没有中断");
                executorService.shutdownNow();
            }
        });
    }

}
